package com.gyt.manager.job;

import com.dangdang.ddframe.job.api.ShardingContext;
import com.dangdang.ddframe.job.api.simple.SimpleJob;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.gyt.manager.common.annotation.ElasticSimpleJob;
import com.gyt.manager.common.utils.DateTimeUtility;
import com.gyt.manager.domain.risk.FunctionLibraryInfo;
import com.gyt.manager.service.FunService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.Delayed;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

@ElasticSimpleJob(cron = "0 37 * * * ?", jobName = "functionTestJob")
@Component
@Slf4j
public class FunctionTestJob implements SimpleJob, InitializingBean, DisposableBean {

    private static final Logger logger = LoggerFactory.getLogger(FunctionTestJob.class);
    @Autowired
    private FunService functionService;

    private final DelayQueue<DelayMessage> delayQueue = new DelayQueue<>();
    private static ExecutorService fixedThreadPool =
            new ThreadPoolExecutor(
                    2,
                    10,
                    10L,
                    TimeUnit.SECONDS,
                    new ArrayBlockingQueue<>(100),
                    new ThreadFactoryBuilder().setNameFormat("run-async-FunctionTestJob-pool-" +
                            RandomStringUtils.randomAlphanumeric(5) + "-%d").build()
            );
    @Override
    public void execute(ShardingContext shardingContext) {
        logger.info("FunctionTestJob execute() ...");
        try {
            Date date = new Date();
            String nowHours = DateTimeUtility.getOneHourAgo(date);
            exeTask(date, nowHours, 5);
        } catch (Throwable e) {
            logger.error("FunctionTestJob error", e);
        } finally {
            logger.info("FunctionTestJob execute end.");
        }
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        try {
            Thread thread = new Thread(() -> {
                while (true) {
                    try {
                        // 从延迟队列的头部获取已经过期的消息
                        // 如果暂时没有过期消息或者队列为空，则take()方法会被阻塞，直到有过期的消息为止
                        DelayMessage delayMessage = delayQueue.take();
                        logger.info("Consumer received message: {}", delayMessage);

                        fixedThreadPool.submit(delayMessage.getMessage());
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
            thread.start();
        } catch (Throwable e) {
            logger.error("FunctionTestJob error", e);
        }
    }

    @Override
    public void destroy() throws Exception {
        fixedThreadPool.shutdown();
    }

    private void exeTask(Date date, String time, int retry){
        try {
            logger.warn("function test job date:{}, time:{},retry:{} start", DateTimeUtility.timeofnow(date),time,retry);
            FunctionLibraryInfo libraryInfo = functionService.findFunctionTypeInfoById(9);
            boolean flag = libraryInfo == null ? false : true;
            long ttl = TimeUnit.MINUTES.toMillis(2);
            if (!flag){
                logger.warn("function test job date:{}, time:{},retry:{} start", DateTimeUtility.timeofnow(date),time,retry);
                retry = retry -1;
                if (retry > 0){
                    Task task = new Task(date, time, retry);
                    logger.warn("-------- offer delay queue", task);
                    delayQueue.offer(DelayMessage.of(task, ttl));
                }
            }
        } catch (Exception e){
            logger.error("function test job error", e);
        }
    }
    public class Task implements Runnable {

        private Date date;

        private String time;

        private int retry = 5;

        public Task(Date date, String time, int retry) {
            this.date = date;
            this.time = time;
            this.retry = retry;
        }
        @Override
        public void run() {
            exeTask(date, time, retry);
        }

        public Date getDate() {
            return date;
        }

        public String getTime() {
            return time;
        }

        public void setDate(Date date) {
            this.date = date;
        }

        public void setTime(String time) {
            this.time = time;
        }

        public int getRetry() {
            return retry;
        }

        public void setRetry(int retry) {
            this.retry = retry;
        }

        @Override
        public String toString() {
            return "Task{" +
                    ", date='" + date + '\'' +
                    ", time='" + time + '\'' +
                    ", retry='" + retry + '\'' +
                    '}';
        }
        public String getTaskId() {
            return date + " " + time;
        }
    }
    public static class DelayMessage implements Delayed {

        private Task message;
        private long ttl;

        /**
         * 构造函数
         * @param message 消息实体
         * @param ttl 延迟时间，单位毫秒
         */
        public static DelayMessage of(Task message, long ttl) {
            DelayMessage delayMessage = new DelayMessage();
            delayMessage.message = message;
            delayMessage.ttl = System.currentTimeMillis() + ttl;
            return delayMessage;
        }

        @Override
        public long getDelay(TimeUnit unit) {
            // 计算该任务距离过期还剩多少时间
            long remaining = ttl - System.currentTimeMillis();
            return unit.convert(remaining, TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(Delayed o) {
            // 比较、排序：对任务的延时大小进行排序，将延时时间最小的任务放到队列头部
            return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
        }

        public Task getMessage() {
            return message;
        }

        public void setMessage(Task message) {
            this.message = message;
        }

        public long getTtl() {
            return ttl;
        }

        public void setTtl(long ttl) {
            this.ttl = ttl;
        }

        @Override
        public String toString() {
            return "DelayMessage{" +
                    "message=" + message +
                    ", ttl=" + ttl +
                    '}';
        }
    }
}
